package dao

import (
	"bytes"
	"context"
	"database/sql"
	"fmt"
	"github.com/google/uuid"
	"github.com/nsqio/go-nsq"
	"github.com/pkg/errors"
	"gorm.io/gorm"
	"mall/internal/conf"
	"mall/internal/constant"
	"mall/internal/model"
	"mall/pkg/ecode"
	"mall/pkg/log"
	"mall/pkg/request_key"
	"time"
)

// 重试NSQ消息
func (d *Dao) RetryNSQMessages(ctx context.Context, msgs []*model.NsqMessage) (err error) {
	if nsqTransactionMsgs, ok := ctx.Value(_nsqTransactionMsgs{}).(map[string]map[uint64][]byte); ok {
		for _, msg := range msgs {
			if _, hasTopic := nsqTransactionMsgs[msg.Topic]; !hasTopic {
				nsqTransactionMsgs[msg.Topic] = make(map[uint64][]byte)
			}
			nsqTransactionMsgs[msg.Topic][msg.ID] = msg.Payload
		}
	} else {
		err = ecode.WithStack(fmt.Errorf("事务消息未初始化"))
		return
	}
	return
}

func (d *Dao) MultiPublishWithoutTransaction(ctx context.Context, topic string, messages [][]byte) (err error) {
	nsqChannel := make(chan *nsq.ProducerTransaction)
	go func() {
		newCtx := request_key.NewContextWithReqID(ctx)
		defer ecode.AsyncFailOver(newCtx, "数据库事务消息发送", conf.Conf.Env, nil)
		res := <-nsqChannel
		if res.Error != nil {
			d.logNsqPublish(ctx, topic, messages, res.Error)
		}
	}()
	if err = d.NSQProducer.MultiPublishAsync(topic, messages, nsqChannel); err != nil {
		d.logNsqPublish(ctx, topic, messages, err)
		return
	}
	return
}

// 发送NSQ消息
func (d *Dao) PublishUniqueNSQMessage(ctx context.Context, topic string, body []byte) (err error) {
	if nsqTransactionMsgs, ok := ctx.Value(_nsqTransactionMsgs{}).(map[string]map[uint64][]byte); ok {
		if _, hasTopic := nsqTransactionMsgs[topic]; !hasTopic {
			nsqTransactionMsgs[topic] = make(map[uint64][]byte)
		}
		// 这里必须重新初始化时间，外部now可能为推送时间，不适合用于建立定时任务
		now := time.Now()
		executeTime := sql.NullTime{Time: now.Add(1 * time.Minute), Valid: true}
		nsqMsg := model.NsqMessage{
			Topic:       topic,
			MsgKey:      uuid.New().String(),
			Payload:     body,
			Status:      constant.NSQInit,
			ExecuteTime: executeTime,
			Retry:       0,
		}
		if err = d.Create(ctx, &nsqMsg); err != nil {
			return
		}
		nsqTransactionMsgs[topic][nsqMsg.ID] = body
	} else {
		err = ecode.WithStack(fmt.Errorf("事务消息未初始化"))
		return
	}
	return
}

// 发送可重复NSQ消息
func (d *Dao) PublishRepeatedNSQMessage(ctx context.Context, topic, msgKey string, body []byte) (err error) {
	if nsqTransactionMsgs, ok := ctx.Value(_nsqTransactionMsgs{}).(map[string]map[uint64][]byte); ok {
		if _, hasTopic := nsqTransactionMsgs[topic]; !hasTopic {
			nsqTransactionMsgs[topic] = make(map[uint64][]byte)
		}
		// 这里必须重新初始化时间，外部now可能为推送时间，不适合用于建立定时任务
		now := time.Now()
		executeTime := sql.NullTime{Time: now.Add(1 * time.Minute), Valid: true}
		nsqMsg := &model.NsqMessage{}
		if err = d.TransContext(ctx).Where(
			"topic = ? and msg_key = ?", topic, msgKey).First(nsqMsg).Error; err != nil {
			err = d.processSqlError(err)
			if errors.Cause(err) != gorm.ErrRecordNotFound {
				return
			}
			nsqMsg = &model.NsqMessage{
				Topic:       topic,
				MsgKey:      msgKey,
				Payload:     body,
				Status:      constant.NSQInit,
				ExecuteTime: executeTime,
				Retry:       0,
			}
			if err = d.Create(ctx, &nsqMsg); err != nil {
				return
			}
		} else {
			nsqMsg.Payload = body
			nsqMsg.Status = constant.NSQInit
			nsqMsg.ExecuteTime = executeTime
			nsqMsg.Retry = 0
			if err = d.Save(ctx, nsqMsg); err != nil {
				return
			}
		}
		nsqTransactionMsgs[topic][nsqMsg.ID] = body
	} else {
		err = ecode.WithStack(fmt.Errorf("事务消息未初始化"))
		return
	}
	return
}

func (d *Dao) logNsqPublish(ctx context.Context, topic string, msgBodies [][]byte, err error) {
	var errString, statusCode string
	if err != nil {
		statusCode = "Fail-Unknown"
		errString = fmt.Sprintf("nsq error: %+v", err)
	} else {
		statusCode = "Success"
	}
	log.FromContext(ctx).Infow("nsq",
		"status_code", statusCode,
		"method", "publish",
		"path", topic,
		"body", string(bytes.Join(msgBodies, []byte(","))),
		"error", errString,
	)
}

func (d *Dao) nsqFailed(ctx context.Context, topic string, msgIds []uint64, msgBodies [][]byte, err error) {
	if dbErr := d.TransContext(ctx).Model(&model.NsqMessage{}).Where(
		"id in ? AND status = ?", msgIds, constant.NSQInit).Updates(map[string]interface{}{
		"retry": gorm.Expr("inv_nsq_message.retry + 1"),
		"execute_time": gorm.Expr(
			"inv_nsq_message.execute_time + interval'1 minute' * inv_nsq_message.retry * inv_nsq_message.retry")}).Error; dbErr != nil {
		// NSQ发送失败时，不抛出错误
		err = d.processSqlError(err)
		log.FromContext(ctx).Error("NSQ Schedule update Failed, err=%s", dbErr.Error())
	}
	d.logNsqPublish(ctx, topic, msgBodies, err)
}

func (d *Dao) nsqSucceeded(ctx context.Context, topic string, msgIds []uint64, msgBodies [][]byte) {
	var (
		err         error
		executeTime *time.Time
	)
	if err = d.TransContext(ctx).Model(&model.NsqMessage{}).Where(
		"id in ? AND status = ?", msgIds, constant.NSQInit).Updates(map[string]interface{}{
		"status":       constant.NSQSucceed,
		"execute_time": executeTime}).Error; err != nil {
		log.FromContext(ctx).Error("NSQ Schedule update Failed, err=%s", err.Error())
		err = d.processSqlError(err)
	}
	// nsq任务更新失败时，同样打失败日志
	d.logNsqPublish(ctx, topic, msgBodies, err)
}
